package com.hgcode.wtboot.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

//@Configuration
@Slf4j
public class MqConsuemrConfig {

	@Value("${xxx.consumer.group:xxxx}")
	private String consumerGroup;

	@Value("${xxx.namesrvAddr:xxxx}")
	private String namesrvAddr;

	@Value("${xxx.namesrvAddr.instanceName:xxxx}")
	private String instanceName;

	@Value("${xxx.consumer.topic:xxxtopic}")
	private String topic;


	@Bean(destroyMethod = "shutdown")
	public DefaultMQPushConsumer buildOrderPushConsumer(OrderConsumer orderConsumer) throws MQClientException {
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
		log.info( "MQ:starting ChronicDiseasePushConsumer ......");
		consumer.setNamesrvAddr(namesrvAddr);
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
		consumer.setMessageModel(MessageModel.CLUSTERING);
		consumer.subscribe(topic, "*");
		consumer.registerMessageListener(orderConsumer);
		consumer.setInstanceName(instanceName);
		consumer.setConsumeThreadMax(1);
		consumer.setConsumeThreadMin(1);
		log.info("MQ：starting buildOrderPushConsumer  is success ! group:{},topic:{},tag:{}", consumerGroup, topic, "*");
		return consumer;
	}
}
